package com.ice.remoting.netty;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.springframework.context.ApplicationContext;

import com.ice.remoting.NettyServerConfig;
import com.ice.remoting.common.RemotingHelper;
import com.ice.remoting.common.ThreadFactoryImpl;
import com.ice.remoting.common.ThreadUtils;
import com.ice.remoting.data.DeviceLiveInfo;
import com.ice.remoting.listener.ChannelDisConnectionEvent;
import com.ice.remoting.utils.SpringContextUtil;

import cn.hutool.core.util.ObjectUtil;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class DeviceManager {

    private final static int DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;

    private ConcurrentHashMap<String, DeviceLiveInfo> liveDeviceInfoMap = new ConcurrentHashMap<>(256);

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    private ApplicationContext applicationContext = SpringContextUtil.applicationContext;
    private NettyServerConfig nettyServerConfig;
    private  ScheduledExecutorService scanExecutorService;


    public DeviceManager(NettyServerConfig nettyServerConfig) {
        this.nettyServerConfig = nettyServerConfig;
    }

    public void initialize(){
        this.scanExecutorService = ThreadUtils.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("DefaultBrokerHeartbeatManager_scheduledService_"));
    }

    /**
     * 监听客户端是否还存活
     *
     * @Author wangwei
     * @Date 2024/1/22
     */
    public void start() {
        this.scanExecutorService.scheduleAtFixedRate(this::scanNotActiveBroker, 5, this.nettyServerConfig.getScanNotActiveBrokerInterval(), TimeUnit.MILLISECONDS);
    }

    public void shutdown(){
        this.scanExecutorService.shutdown();
    }

    public void putDeviceLive(String key, Integer timeoutMillis, Channel channel) {
        try {
            this.lock.writeLock().lockInterruptibly();
            DeviceLiveInfo deviceLiveInfoResult = liveDeviceInfoMap.put(key, new DeviceLiveInfo(
                    System.currentTimeMillis(),
                    timeoutMillis == null ? DEFAULT_BROKER_CHANNEL_EXPIRED_TIME : timeoutMillis,
                    channel));
            if (deviceLiveInfoResult == null) {
                log.info("new broker registered, {} ", key);
            }
        } catch (Exception e) {
            log.error("putDeviceLive Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public DeviceLiveInfo getDeviceChannel(Channel channel) {
        DeviceLiveInfo deviceLiveFound = null;
        for (Map.Entry<String, DeviceLiveInfo> entry : this.liveDeviceInfoMap.entrySet()) {
            if (entry.getValue().getChannel() == channel) {
                deviceLiveFound = entry.getValue();
                break;
            }
        }
        return deviceLiveFound;
    }

    public DeviceLiveInfo getDeviceInfo(String customerUniqueIdentity) {
        if (ObjectUtil.isEmpty(customerUniqueIdentity)) {
            return null;
        }
        DeviceLiveInfo deviceLiveFound = null;
        for (Map.Entry<String, DeviceLiveInfo> entry : this.liveDeviceInfoMap.entrySet()) {
            if (entry.getValue().getCustomerUniqueIdentity().equals(customerUniqueIdentity)) {
                deviceLiveFound = entry.getValue();
                break;
            }
        }
        return deviceLiveFound;
    }

    public void putDeviceLive(String key, DeviceLiveInfo deviceLiveInfo) {
        try {
            this.lock.writeLock().lockInterruptibly();
            DeviceLiveInfo deviceLiveInfoResult = liveDeviceInfoMap.put(key, deviceLiveInfo);
            if (deviceLiveInfoResult == null) {
                log.info("new broker registered, {} ", key);
            }
        } catch (Exception e) {
            log.error("putDeviceLive Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public void setCustomerUniqueIdentity(Channel channel, String customerUniqueIdentity) {
        DeviceLiveInfo deviceLiveInfo = this.getDeviceChannel(channel);
        if (deviceLiveInfo != null) {
            deviceLiveInfo.setCustomerUniqueIdentity(customerUniqueIdentity);
            return;
        }
        log.info("channel已关闭,{}", RemotingHelper.parseChannelRemoteAddr(channel));
    }

    public void remove(Channel channel) {
        try {
            String deviceLiveFound = null;
            this.lock.writeLock().lockInterruptibly();
            for (Map.Entry<String, DeviceLiveInfo> entry : this.liveDeviceInfoMap.entrySet()) {
                if (entry.getValue().getChannel() == channel) {
                    deviceLiveFound = entry.getKey();
                    break;
                }
            }
            if (deviceLiveFound != null) {
                DeviceLiveInfo oldDeviceInfo = this.liveDeviceInfoMap.remove(deviceLiveFound);
                log.info("unregisterBroker, remove from brokerLiveTable {}, {}", oldDeviceInfo != null ? "OK" : "Failed", deviceLiveFound);
            }
        } catch (Exception e) {
            log.error("device remove Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public void updateBrokerInfoUpdateTimestamp(final String key) {
        DeviceLiveInfo prev = this.liveDeviceInfoMap.get(key);
        if (prev != null) {
            prev.setLastUpdateTimestamp(System.currentTimeMillis());
        }
    }

    /**
     * 不走加锁设置，删除不活跃的可以不那么准时下次扫描不影响
     *
     * @Author wangwei
     * @Date 2024/1/23
     */
    public void scanNotActiveBroker() {
        try {
            log.info("start scanNotActiveBroker");
            final Iterator<Map.Entry<String, DeviceLiveInfo>> iterator = this.liveDeviceInfoMap.entrySet().iterator();
            while (iterator.hasNext()) {
                final Map.Entry<String, DeviceLiveInfo> next = iterator.next();
                long last = next.getValue().getLastUpdateTimestamp();
                long timeoutMillis = next.getValue().getHeartbeatTimeoutMillis();
                if ((last + timeoutMillis) < System.currentTimeMillis()) {
                    Channel channel = next.getValue().getChannel();
                    iterator.remove();
                    if (channel != null) {
                        RemotingHelper.closeChannel(channel);
                    }
                    // channel 关闭事件通知
                    ChannelDisConnectionEvent channelDisConnectionEvent = new ChannelDisConnectionEvent(this, channel);
                    if (ObjectUtil.isNotEmpty(applicationContext)) {
                        applicationContext.publishEvent(channelDisConnectionEvent);
                    }
                    log.warn("The broker channel expired, {} {}ms", next.getKey(), timeoutMillis);
                }
            }
        } catch (Exception e) {
            log.error("scanNotActiveBroker exception", e);
        }
    }
}
